Linux死锁 && 生产者与消费者模型 && 信号量

Posted 2023-02-28 Suk-god

文章目录

• 1、死锁
• • 1.1 死锁的产生场景
  • 1.2 死锁的gdb分析
  • 1.3 死锁的必要条件
  • 1.4 死锁的预防
• 2、生产者与消费者模型
• • 2.1 123规则
  • 2.2 应用场景 && 优点
  • 2.4 代码实现（互斥锁+条件变量）
• 3、信号量
• • 3.1 信号量的原理
  • 3.2 信号量的接口
  • • 3.2.1 初始化接口
    • 3.2.2 等待接口
    • 3.2.3 释放接口
    • 3.2.4 销毁接口
  • 3.3 两点注意事项
  • 3.4 使用信号量实现生产者与消费者模型

1、死锁

1.1 死锁的产生场景

1、线程加锁之后，没有释放互斥锁就退出了

模拟实现

分析结果

因此，我们要在线程所有可能退出的地方都释放互斥锁

2、两种线程分别拿着一把锁，还想要请求对方的锁

模拟实现：

分析结果：

1.2 死锁的gdb分析

产生了死锁，那应该如何去分析死锁产生的原因呢？
我们可以通过gdb来调试该程序，进而判断出死锁产生的原因.
我们接下来就以上面死锁产生情况2的代码为例，采用下面的两种方式分别进行调试！

1、通过调试可执行程序来分析

方式：打断点 + 调试

1、b + 行号 打断点

2、使用thread apply all bt 命令，将所有线程的调用堆栈情况都展现出来

3、使用p + 互斥锁变量可以查看互斥锁的内容。

2、通过调试正在运行的程序

方式：gdb attach + 进程号

前提：程序正在运行，并且能够获取到进程的进程号

此时，就进入了gdb调试当中，我们可以按照上述查看线程的调用堆栈以及查看互斥锁内容等命令来分析我们的代码

1.3 死锁的必要条件

1. 不可剥夺
   线程获取到互斥锁后，除了自己释放，其他线程不能进行释放
2. 循环等待
   线程A拿着1锁请求2锁，线程B拿着2锁请求1锁
3. 互斥条件
   一个互斥锁在同一时间只能被一个线程拥有
4. 请求与保持
   吃着碗里的，看着锅里的，其实就是死锁产生的情况2那种情形

1.4 死锁的预防

1. 破坏必要条件
   
2. 加锁顺序一致
   都先加同一把锁，再去加另外一把锁。
3. 避免锁没有被释放
   在线程所有可能退出的地方都进行解锁
4. 资源一次性分配
   

2、生产者与消费者模型

2.1 123规则

1---->1个线程安全的队列
2---->两种角色，生产者和消费者
3---->3个规则
  生产者与生产者互斥
  消费者与消费者互斥
  生产者与消费者 互斥 + 同步

2.2 应用场景 && 优点

一般应用于后端程序当中

画图说明：

优点：
1、忙闲不均
2、生产者与消费者高度解耦
3、支持高并发

2.4 代码实现（互斥锁+条件变量）

#include<stdio.h>
#include<unistd.h>
#include<pthread.h>
#include<queue>

using namespace std;

#define THREAD_COUNT 2


//创建一个线程安全队列
class RingQueue

public:
    RingQueue()
    
        capacity_ = 1;
        //初始化互斥锁
        pthread_mutex_init(&que_lock_,NULL);
        //初始化条件变量
        pthread_cond_init(&cons_cond_,NULL);
        pthread_cond_init(&prod_cond_,NULL);
    
    ~RingQueue()
    
        //销毁互斥锁
        pthread_mutex_destroy(&que_lock_);
        //销毁条件变量
        pthread_cond_destroy(&cons_cond_);
        pthread_cond_destroy(&prod_cond_);
    
    void Push(int data)
    
        //加锁
        pthread_mutex_lock(&que_lock_);
        while(que_.size() >= capacity_)
        
            pthread_cond_wait(&prod_cond_,&que_lock_);
        
        que_.push(data);
        printf("I am prodece thread %p: I produce %d\\n",pthread_self(),data);
        //解锁
        pthread_mutex_unlock(&que_lock_);
        //通知消费者消费
        pthread_cond_signal(&cons_cond_);
    
    void Pop(int* data)
    
        //加锁
        pthread_mutex_lock(&que_lock_);
        while(que_.size() <= 0)
        
            pthread_cond_wait(&cons_cond_,&que_lock_);
        
        *data = que_.front();
        que_.pop();
        printf("I am consume thread %p: I consume %d\\n",pthread_self(),*data);
        //解锁
        pthread_mutex_unlock(&que_lock_);
        //通知生产者生产
        pthread_cond_signal(&prod_cond_);
    

private:
    queue<int> que_;
    size_t capacity_;
    //互斥锁
    pthread_mutex_t que_lock_;

    //同步
    pthread_cond_t cons_cond_;
    pthread_cond_t prod_cond_;
;

int g_data = 0;
//注意：静态初始化互斥锁保证prod_thread_start中临界区的原子性
//不能在该函数内定义互斥锁变量，因为那样多个生产者拿到的互斥锁不是同一个，会产生程序的二义性结果！
pthread_mutex_t g_data_lock = PTHREAD_MUTEX_INITIALIZER;

void* prod_thread_start(void* arg)

    RingQueue* rq = (RingQueue*)arg;
    while(1)
    
        pthread_mutex_lock(&g_data_lock);
        rq->Push(g_data);
        g_data++;
        //usleep(20);
        pthread_mutex_unlock(&g_data_lock);
    


void* cons_thread_start(void* arg)

    RingQueue* rq = (RingQueue*)arg;
    while(1)
    
        int data;
        rq->Pop(&data);
    


int main()

    //实例化一个队列
    RingQueue *rq = new RingQueue();
    //创建生产者和消费者线程
    pthread_t prod[THREAD_COUNT],cons[THREAD_COUNT];
    for(int i = 0; i < THREAD_COUNT; i++)
    
        int ret = pthread_create(&prod[i],NULL,prod_thread_start,(void*)rq);
        if(ret < 0)
        
            perror("pthread_create");
            return 0;
        
        ret = pthread_create(&cons[i],NULL,cons_thread_start,(void*)rq);
        if(ret < 0)
        
            perror("pthread_create");
            return 0;
        
    

    //主线程等待回收工作线程
    for(int i = 0; i < THREAD_COUNT; i++)
    
        pthread_join(prod[i],NULL);
        pthread_join(cons[i],NULL);
    
    delete rq;
    return 0;

输出结果：

3、信号量

3.1 信号量的原理

信号量是由一个资源计数器和一个PCB等待队列构成

PCB等待队列与条件变量实现同步中的PCB是一样的。

对于资源计数器，它的作用与功能如下：

图解：

生产者信号量初始化时资源计数器的值一般为线程安全队列的容量大小
消费者信号量初始化时资源计数器的值一般为0，因为刚开始，线程安全队列中并没有资源可以使用。

3.2 信号量的接口

3.2.1 初始化接口

int sem_init(sem_t * sem, int pshared, int value);

3.2.2 等待接口

int sem_wait(sem_t* sem);

3.2.3 释放接口

int sem_post(sem_t *sem);

3.2.4 销毁接口

int sem_destroy(sem_t* sem);
信号量是动态的初始化的，因此需要销毁。

3.3 两点注意事项

1、对于生产者与消费者来说，获取信号量与加锁的先后顺序是怎样的？

设想一：先拿到互斥锁，再获取信号量
设想二：先获取信号量，再拿互斥锁

究竟谁对谁错？我们逐一来分析

前提：有一个如下图介绍的生产者与消费者模型，上述的两种设想均是按照该模型进行分析的

设想一：先拿互斥锁，再获取信号量
对于这种设想，我们分析一下：

设想一错误！排除

分析设想二：先获取信号量，再获取互斥锁

因此，我们最后得到的结论就是：
 先获取信号量，再保证互斥（方式：互斥锁 || 信号量）

2、信号量既可以保证同步，也可以保证互斥

信号量保证同步，在分析上一个问题的时候已经体现出来了！
这里只分析信号量如何保证互斥的！

其实很简单，只需要将信号量中资源计数器的初始值设置为1，就能够保证互斥了。

3.4 使用信号量实现生产者与消费者模型

1. 思路
   使用一个环形队列作为线程安全队列，该环形队列使用一个数组来模拟。具体的图解如下：
   
2. 代码实现

#include<stdio.h>
#include<unistd.h>
#include<pthread.h>
#include<semaphore.h>

#include<vector>

using namespace std;
#define CAPACITY 2
#define THREAD_COUNT 2

class RingQueue

public:
    RingQueue()
        :vec_(CAPACITY)
        ,capacity_(CAPACITY)
        ,write_pos_(0)
        ,read_pos_(0)
    
        //初始化互斥信号量
        sem_init(&sem_lock_,0,1);
        //初始化同步信号量
        sem_init(&cons_sem_,0,0);
        sem_init(&prod_sem_,0,capacity_);
    

    ~RingQueue()
    
        //销毁互斥信号量
        sem_destroy(&sem_lock_);
        //销毁同步信号量
        sem_destroy(&cons_sem_);
        sem_destroy(&prod_sem_);
    

    void Push(int data)
    
        //获取生产者的信号量
        sem_wait(&prod_sem_);
        //获取互斥锁，保证互斥访问临界区
        sem_wait(&sem_lock_);
        printf("I am produce thread %p: I produce %d\\n",pthread_self(),data);
        vec_[write_pos_] = data;
        write_pos_ = (write_pos_+1)%capacity_;
        sem_post(&sem_lock_);
        //通知消费者进行消费
        sem_post(&cons_sem_);
    

    void Pop()
    
        //获取消费者的信号量
        sem_wait(&cons_sem_);

        sem_wait(&sem_lock_);
        int data = vec_[read_pos_];
        read_pos_ = (read_pos_ + 1) % capacity_;
        printf("I am consume thread %p: I consume %d\\n",pthread_self(),data);
        sem_post(&sem_lock_);
        //通知生产者生产
        sem_post(&prod_sem_);
    
private:
    vector<int> vec_;
    size_t capacity_;

    //保证互斥的信号量
    sem_t sem_lock_;
    //保证同步的信号量
    sem_t cons_sem_;
    sem_t prod_sem_;

    int write_pos_;
    int read_pos_;
;

int g_data = 0;
sem_t g_sem_lock;

void* prod_thread_start(void* arg)

    RingQueue* rq = (RingQueue*) arg;
    while(1)
    
        sem_wait(&g_sem_lock);
        rq->Push(g_data);
        g_data++;
        sem_post(&g_sem_lock);
    


void* cons_thread_start(void* arg)

    RingQueue* rq = (RingQueue*)arg;

    while(1)
    
        rq->Pop();
    


int main()

    RingQueue* rq = new RingQueue();
    //初始化互斥信号量g_sem_lock
    //用于保证多个生产者互斥访问prod_thread_start接口中的临界资源
    sem_init(&g_sem_lock,0,1);

    //创建两类线程，生产者 && 消费者
    pthread_t prod_thread[THREAD_COUNT],cons_thread[THREAD_COUNT];
    for(int i = 0; i < THREAD_COUNT; i++)
    
        int ret = pthread_create(&prod_thread[i],NULL,prod_thread_start,(void*)rq);
        if(ret < 0)
        
            perror("pthread_create");
            return 0;
        
        ret = pthread_create(&cons_thread[i],NULL,cons_thread_start,(void*)rq);
        if(ret < 0)
        
            perror("pthread_create");
            return 0;
        
    

    //主线程等待接收工作线程
    for(int i = 0; i < THREAD_COUNT; i++)
    
        pthread_join(prod_thread[i],NULL);
        pthread_join(cons_thread[i],NULL);
    

    delete rq;
    return 0;

到这里，有关死锁、生产者与消费者模型、信号量等知识总结完毕。感觉有所收获的读友们，欢迎评论转发，分享给身边的朋友！！