进程与线程IPC-信号量（实现生产者消费者&哲学家就餐）

叁歲伎倆 2022-07-28 01:17 119阅读 0赞

#### 进程&线程IPC ####

我们知道操作系统中的很重要的一个是线程&进程，而进程通信通常解决方案也适用于线程，反过来也一样，到线程中就是临界区的互斥访问，因此以下我们只讨论线程。

#### 方法 ####

*  忙等 (优先级反转问题导致死锁)  
    
    
     *  禁止中断  
        
        
         *  权利交给用户线程不明智
         *  对于多CPU 无法处理
     *  锁 读取与加锁不一致导致两个进程同时访问
     *  自旋锁 相互制约
     *  Peterson解法
     *  TSL指令
 *  阻塞等  
    
    
     *  sleep() & wakeup() 唤醒等待位
     *  信号量mutex (多cpu有各自私有内存不支持)
     *  管程  
        
        
         *  一个时刻一个线程运行
         *  条件变量（wait&signal）
         *  语言、编译器支持（java）
     *  消息传递  
        
        
         *  类似于网络
         *  支持多CPU各自有私有内存
     *  屏障

#### 信号量 ####

信号量有两个基本操作**down**和**up**，当然这两个内部都是原子操作，否则就和sleep&wakeup一样出现问题。

*  down  
    
    
     *  >0 减一
     *  ==0 阻塞
 *  up 增1，唤醒

#### 问题 ####

1：像python这样存在GIL线程锁的语言多线程互斥访问临界区为什么还需要加锁？  
因为存在这样的情况，比如两个线程对一个共享变量tmp进行+1操作，首先，线程A读取变量tmp = 0，然后线程B获取CPU，读取tmp=0，然后进行+1，则tmp=1，然后切换到线程A，线程A在之前读取的基础上+1,tmp=1,这样加两次结果还是1，明显不对，所以需要在操作这样的互斥变量的时候加锁

2：多CPU下一个线程的多个线程可以并行运行吗？  
Linux2.4以前不支持线程，只支持轻量级进程，2.4以后重写了线程库，支持了。对于内核级线程，即fork出来的，包括pthread\_create，都是内核级线程，支持多调度CPU进行计算，而用户级进程不支持的，这也是用户级线程的弊端。

#### 经典IPC问题 ####

**生产者消费者**  
这里两个主要问题是同步信号量（empty&full）以及互斥信号量（mutex），在考虑用信号量解决问题的时候，首先考虑使用哪一种。这里提供一个多生产者多消费者的实现版本。

#include <stdio.h>                                                                                                                                                   
    #include <stdlib.h>
    #include <unistd.h>
    #include <pthread.h>
    #include <semaphore.h>
    #include <signal.h>
    #include <time.h>
    
    const int N = 5;
    const int M = 10; //缓冲池数目
    
    int in = 0,out = 0;
    
    int buff[M];
    
    sem_t empty_sem;
    sem_t full_sem;
    pthread_mutex_t mutex;
    
    int product_id = 0;
    int prochase_id = 0;
    
    double random_num()
    {
        return (double)rand() / RAND_MAX;
    }
    int random(int x) //随机数生成算法 (0--x)
    {
        return (int)(random_num()*(x-1)+0.5);
    }
    
    void Handlesignal(int signo) //Ctrl+C
    {
        printf("程序退出 %d !\n", signo);
        exit(0);
    }
    void print()
    {
        for(int i=0; i<M; ++i)
            printf("%d ", buff[i]);
        puts("");
    }
    void *product(void *arg)
    {
        int id = ++product_id;
        while(true)
        {
            sleep( random(4) );
    
            sem_wait(&empty_sem); //down
            pthread_mutex_lock(&mutex);
    
            in = in%M;
            printf("生产者 %d 在产品队列中放入第 %d 个产品\n", id, in);
            buff[in] = 1;
            print();
            ++in;
    
            pthread_mutex_unlock(&mutex);
            sem_post(&full_sem);
        }
    }
    void* prochase(void *arg) //记得传入加个参数
    {
        int id = ++prochase_id;
        while(true) ///
        {
            int tmp = random(4);
            printf("sheng* %d *****************************************************************************\n", tmp);
    
            sleep( tmp );
    
            sem_wait(&full_sem);
            pthread_mutex_lock(&mutex);
    
            out = out%M;
            printf("消费者 %d 在产品队列中消费了第 %d 个产品\n", id, out);
            buff[out] = 0;
            print();
            ++out;
    
            pthread_mutex_unlock(&mutex);
            sem_post(&empty_sem);
        }
    }
    int main()
    {
        if(signal(SIGINT, Handlesignal) == SIG_ERR)
        {
            printf("信号量出错！");
        }
        //初始化信号量
        int init1 = sem_init(&empty_sem, 0, M);
        int init2 = sem_init(&full_sem, 0, 0);
    
        int init3 = pthread_mutex_init(&mutex, NULL);
        if(init1 || init2 || init3)
        {
            printf("信号量初始化失败！\n");
            exit(1);
        }
        int ret[N];
        pthread_t id1[N];
        pthread_t id2[N];
        for(int i=0; i<N; ++i)
        {
            ret[i]= pthread_create(&id1[i], NULL, product, NULL);
            if(ret[i])
            {
                printf("生产者 %d 创建失败！\n", i);
                exit(1);
            }
        }
        for(int i=0; i<N; ++i)
        {
            ret[i] = pthread_create(&id2[i], NULL, prochase, NULL);
            if(ret[i])
            {
                printf("消费者 %d 创建失败!\n",i);
                exit(1);
            }
        }
        for(int i=0; i<N; ++i)
        {
            pthread_join(id1[i], NULL);
            pthread_join(id2[i], NULL);
        }
        exit(0);
    }

**哲学家就餐问题**  
这个是经典IPC中比较难的一个问题，这里的实现能够实现最大并行度的，在一个哲学家用餐结束的时候如果可以会唤醒相邻两个哲学家就餐。

#include <stdio.h>                                                                                                                                                   
    #include <stdlib.h>
    #include <unistd.h>
    #include <pthread.h>
    #include <semaphore.h>
    #include <signal.h>
    #include <time.h>
    #include <string>
    
    using namespace std;
    #define THINK 0
    #define HUNGRY 1
    #define EAT 2
    const int N = 5; //哲学家数目
    
    pthread_mutex_t mutex;
    pthread_mutex_t out;
    pthread_mutex_t s[7];
    
    int state[N];
    
    double random_num()
    {
        return (double)rand() / RAND_MAX;
    }
    int random(int x) //随机数生成算法 (0--x)
    {
        return (int)(random_num()*(x-1)+0.5);
    }
    
    void Handlesignal(int signo) //Ctrl+C
    {
        printf("程序退出 %d !\n", signo);
        exit(0);
    }
    void print()
    {
        //for(int i=0;i<N; ++i)
        //    printf(" %d",i);
        //    这一块儿需要加锁
        pthread_mutex_lock(&out);
        for(int i=0;i<N; ++i)
        { 
            if(state[i] == THINK)
                printf("THINK ");
            else if(state[i] == EAT)
                printf("EAT ");
            else
                printf("HUNGRY ");
        }
        puts("");
        pthread_mutex_unlock(&out);
    }
    void test(int i)
    {
        if(state[i] == HUNGRY && state[ (i+4)%5 ] != EAT && state[ (i+1)%5 ] !=EAT)
        {
            state[i] = EAT;
            pthread_mutex_unlock(&s[i]);
        }
    }
    
    void take_forks(int i)
    {
        pthread_mutex_lock(&mutex);
        state[i] = HUNGRY;
        test(i);
        pthread_mutex_unlock(&mutex);
        pthread_mutex_lock(&s[i]);
    
    }
    void put_forks(int i)
    {
        pthread_mutex_lock(&mutex);
        state[i] = THINK;
        test( (i+4)%5 );
        test( (i+1)%5 );
        pthread_mutex_unlock(&mutex);
    }
    
    
    void* think(void *arg)
    {
        int i = *(int *)arg;
        while(true)
        {
            sleep( random(4) ); //think
            printf("第 %d 个哲学家在思考!\n", i);
            take_forks(i);
            print();
            sleep( random(4) ); //eat
            printf("第 %d 个哲学家在吃饭！\n", i);
            put_forks(i);
        }
    } 
    
    int main()
    {
        if(signal(SIGINT, Handlesignal) == SIG_ERR)
            printf("信号量出错!\n");
        int init1 = pthread_mutex_init(&mutex, NULL);
        int init2 = pthread_mutex_init(&out, NULL);
        if(init1 != 0 || init2 != 0)
        {
            printf("线程mutex初始化失败！\n");
            exit(1);
        }
        int ret[N];
        pthread_t id[N+2];
        for(int i=0; i<N; ++i)
        {
            int init = pthread_mutex_init(&s[i], NULL);
            if (init != 0)
                printf("哲学家 %d 信号量初始化失败！\n", s[i]);
        }
        for(int i=0; i<N; ++i)
        {
            ret[i] = pthread_create(&id[i], NULL, think, &i);
            if(ret[i] != 0)
                printf("哲学家 %d 初始化失败！\n", i);
        }
        for(int i=0; i<N; ++i)
            pthread_join(id[i], NULL);
    
        exit(0);
    }